package com.navinfo.opentsp.platform.computing.serializer;

import com.navinfo.opentsp.platform.location.protocol.common.LCObdInformation;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.AsyncHbaseEventSerializer;
import org.apache.hadoop.hbase.util.Bytes;
import org.hbase.async.AtomicIncrementRequest;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * OBD数据hbase落盘
 */
public class AsyncHbaseObdDataSerializer implements AsyncHbaseEventSerializer {

    private static final Logger log = LoggerFactory.getLogger(AsyncHbaseObdDataSerializer.class);

    private byte[] table;
    private byte[] colFam =
            Bytes.toBytes("d");
    private Event currentEvent;

    @Override
    public void initialize(byte[] table, byte[] colFam) {
        this.table = table;
        this.colFam = colFam;
    }

    @Override
    public void setEvent(Event event) {
        this.currentEvent = event;
    }

    @Override
    public List<PutRequest> getActions() {
        List<PutRequest> actions = new ArrayList();
        try {
            LCObdInformation.ObdInformation obdInformation = LCObdInformation.ObdInformation.parseFrom(currentEvent.getBody());
            //rowkey:reverse(tId)|3000000000L-gpsTime
            //reverse(tId)实现rowkey的散列，3000000000L-gpsTime实现逆序最新记录在头部
            String rowKey = new StringBuilder(Long.valueOf(obdInformation.getTerminalId()).toString()).reverse() + "|" + (3000000000L - obdInformation.getGpsTime());

            //添加PutRequest
            actions.add(preparedata(rowKey , obdInformation));
        } catch (Exception e) {
            log.error("LocationData异步序列化失败：" + e.getMessage(), e);
        }

        return actions;
    }

    /**
     * 处理数据
     * @param rowKey
     *      rowKey
     * @param obdInformation
     *      obdInformation
     * @throws Exception
     *      Exception
     */
    private PutRequest preparedata(String rowKey, LCObdInformation.ObdInformation obdInformation) throws Exception {
        byte[][] qualifiers = new byte[24][];
        byte[][] values = new byte[24][];

        qualifiers[0] = Bytes.toBytes("protocolType");
        qualifiers[1] = Bytes.toBytes("milType");
        qualifiers[2] = Bytes.toBytes("diagnosticSupportStatus");
        qualifiers[3] = Bytes.toBytes("diagnosticReadinessStatus");
        qualifiers[4] = Bytes.toBytes("calid");
        qualifiers[5] = Bytes.toBytes("vin");
        qualifiers[6] = Bytes.toBytes("calibrationIdentifyNO");
        qualifiers[7] = Bytes.toBytes("cvn");
        qualifiers[8] = Bytes.toBytes("iupr");
        qualifiers[9] = Bytes.toBytes("faultCount");
        qualifiers[10] = Bytes.toBytes("faultCodes");
        qualifiers[11] = Bytes.toBytes("isPatch");
        qualifiers[12] = Bytes.toBytes("gpsTime");
        qualifiers[13] = Bytes.toBytes("encrypt");
        qualifiers[14] = Bytes.toBytes("terminalVersion");
        qualifiers[15] = Bytes.toBytes("terminalId");
        qualifiers[16] = Bytes.toBytes("serialNumber");
        qualifiers[17] = Bytes.toBytes("rLength");
        qualifiers[18] = Bytes.toBytes("sLength");
        qualifiers[19] = Bytes.toBytes("rValue");
        qualifiers[20] = Bytes.toBytes("sValue");
        qualifiers[21] = Bytes.toBytes("originalPB");
//        qualifiers[22] = Bytes.toBytes("spnFmi");
//        qualifiers[23] = Bytes.toBytes("pCode");

        values[0] = Bytes.toBytes(String.valueOf(obdInformation.getProtocolType()));
        values[1] = Bytes.toBytes(String.valueOf(obdInformation.getMilType()));
        values[2] = Bytes.toBytes(String.valueOf(obdInformation.getDiagnosticSupportStatus()));
        values[3] = Bytes.toBytes(String.valueOf(obdInformation.getDiagnosticReadinessStatus()));
        values[4] = Bytes.toBytes(String.valueOf(obdInformation.getCalid()));
        values[5] = Bytes.toBytes(String.valueOf(obdInformation.getVin()));
        values[6] = Bytes.toBytes(String.valueOf(obdInformation.getCalibrationIdentifyNO()));
        values[7] = Bytes.toBytes(String.valueOf(obdInformation.getCvn()));
        values[8] = Bytes.toBytes(String.valueOf(obdInformation.getIuprList()));
        values[9] = Bytes.toBytes(String.valueOf(obdInformation.getFaultCount()));
        values[10] = Bytes.toBytes(String.valueOf(obdInformation.getFaultCodesList()));
        values[11] = Bytes.toBytes(String.valueOf(obdInformation.getIsPatch()));
        values[12] = Bytes.toBytes(String.valueOf(obdInformation.getGpsTime()));
        values[13] = Bytes.toBytes(String.valueOf(obdInformation.getEncrypt()));
        values[14] = Bytes.toBytes(String.valueOf(obdInformation.getTerminalVersion()));
        values[15] = Bytes.toBytes(String.valueOf(obdInformation.getTerminalId()));
        values[16] = Bytes.toBytes(String.valueOf(obdInformation.getSerialNumber()));
        values[17] = Bytes.toBytes(String.valueOf(obdInformation.getRLength()));
        values[18] = Bytes.toBytes(String.valueOf(obdInformation.getSLength()));
        values[19] = Bytes.toBytes(String.valueOf(obdInformation.getRValue()));
        values[20] = Bytes.toBytes(String.valueOf(obdInformation.getSValue()));
        values[21] = obdInformation.toByteArray();
//        values[22] = Bytes.toBytes(String.valueOf(obdInformation.getSpnFmiFaultCodeList()));
//        values[23] = Bytes.toBytes(String.valueOf(obdInformation.getPcodeFaultCodecList()));
        return new PutRequest(table, Bytes.toBytes(rowKey), colFam, qualifiers, values);
    }


    @Override
    public List<AtomicIncrementRequest> getIncrements() {
        return new ArrayList<AtomicIncrementRequest>();
    }

    @Override
    public void configure(Context context) {
    }
    @Override
    public void configure(ComponentConfiguration componentConfiguration) {
    }
    @Override
    public void cleanUp() {    }
}
